package com.taotao.cannal;


import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author 余胜军
 * @ClassName CanalConsumer
 * @qq 644064779
 * @addres www.mayikt.com
 * @createTime 2021年04月27日 19:14:00
 */
@Component
@Slf4j
public class CanalConsumer {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 消费者监听mayikt-topic
     *
     * @param consumer
     */
    @KafkaListener(topics = "mayikt-topic")
    public void receive01(ConsumerRecord<?, String> consumer) {
        log.info("分组1的消费者1>topic名称:{},,key:{},分区位置:{},offset{},数据:{}<",
                consumer.topic(), consumer.key(), consumer.partition(), consumer.offset(), consumer.value());
        JSONObject jsonObject = JSONObject.parseObject(consumer.value());
        String type = jsonObject.getString("type");
        switch (type) {
            case MayiktConstant.CANAL_INSERT:
            case MayiktConstant.CANAL_UPDATE:
                updateRedis(jsonObject);
                return;
            case MayiktConstant.CANAL_DELETE:
                deleteRedis(jsonObject);
        }
    }

    /**
     * 同步到Redis
     *
     * @param jsonObject
     */
    private void updateRedis(JSONObject jsonObject) {
        log.info("updateRedis:{}", jsonObject);
        JSONArray datas = jsonObject.getJSONArray("data");
        for (int i = 0; i < datas.size(); i++) {
            JSONObject info = datas.getJSONObject(i);
            String id = info.getString("id");
            redisTemplate.opsForValue().set(id, info.toJSONString());
        }
    }

    private void deleteRedis(JSONObject jsonObject) {
        log.info("deleteRedis:{}", jsonObject);
        JSONArray datas = jsonObject.getJSONArray("data");
        for (int i = 0; i < datas.size(); i++) {
            JSONObject info = datas.getJSONObject(i);
            String id = info.getString("id");
            redisTemplate.delete(id);
        }
    }
}
